总数|单名_大数据中必须要掌握的FlinkSQL详细剖析

篇首语：本文由编程笔记#小编为大家整理，主要介绍了大数据中必须要掌握的 Flink SQL 详细剖析相关的知识，希望对你有一定的参考价值。

目录

1. Flink SQL 常用算子

2. Flink SQL 实战案例

Flink SQL 是 Flink 实时计算为简化计算模型&＃xff0c;降低用户使用实时计算门槛而设计的一套符合标准 SQL 语义的开发语言。 自 2015 年开始&＃xff0c;阿里巴巴开始调研开源流计算引擎&＃xff0c;最终决定基于 Flink 打造新一代计算引擎&＃xff0c;针对 Flink 存在的不足进行优化和改进&＃xff0c;并且在 2019 年初将最终代码开源&＃xff0c;也就是我们熟知的 Blink。Blink 在原来的 Flink 基础上最显著的一个贡献就是 Flink SQL 的实现。

Flink SQL 是面向用户的 API 层&＃xff0c;在我们传统的流式计算领域&＃xff0c;比如 Storm、Spark Streaming 都会提供一些 Function 或者 Datastream API&＃xff0c;用户通过 Java 或 Scala 写业务逻辑&＃xff0c;这种方式虽然灵活&＃xff0c;但有一些不足&＃xff0c;比如具备一定门槛且调优较难&＃xff0c;随着版本的不断更新&＃xff0c;API 也出现了很多不兼容的地方。

在这个背景下&＃xff0c;毫无疑问&＃xff0c;SQL 就成了我们最佳选择&＃xff0c;之所以选择将 SQL 作为核心 API&＃xff0c;是因为其具有几个非常重要的特点&＃xff1a;

• SQL 属于设定式语言&＃xff0c;用户只要表达清楚需求即可&＃xff0c;不需要了解具体做法&＃xff1b;

• SQL 可优化&＃xff0c;内置多种查询优化器&＃xff0c;这些查询优化器可为 SQL 翻译出最优执行计划&＃xff1b;

• SQL 易于理解&＃xff0c;不同行业和领域的人都懂&＃xff0c;学习成本较低&＃xff1b;

• SQL 非常稳定&＃xff0c;在数据库 30 多年的历史中&＃xff0c;SQL 本身变化较少&＃xff1b;

• 流与批的统一&＃xff0c;Flink 底层 Runtime 本身就是一个流与批统一的引擎&＃xff0c;而 SQL 可以做到 API 层的流与批统一。

1. Flink SQL 常用算子

SELECT&＃xff1a;

SELECT 用于从 DataSet/DataStream 中选择数据&＃xff0c;用于筛选出某些列。

示例&＃xff1a;

SELECT * FROM Table; // 取出表中的所有列

SELECT name&＃xff0c;age FROM Table; // 取出表中 name 和 age 两列

与此同时 SELECT 语句中可以使用函数和别名&＃xff0c;例如我们上面提到的 WordCount 中&＃xff1a;

SELECT word, COUNT(word) FROM table GROUP BY word;

WHERE&＃xff1a;

WHERE 用于从数据集/流中过滤数据&＃xff0c;与 SELECT 一起使用&＃xff0c;用于根据某些条件对关系做水平分割&＃xff0c;即选择符合条件的记录。

示例&＃xff1a;

SELECT name&＃xff0c;age FROM Table where name LIKE ‘% 小明 %’;

SELECT * FROM Table WHERE age &＃61; 20;

WHERE 是从原数据中进行过滤&＃xff0c;那么在 WHERE 条件中&＃xff0c;Flink SQL 同样支持 &＃61;、<、>、<>、>&＃61;、<&＃61;&＃xff0c;以及 AND、OR 等表达式的组合&＃xff0c;最终满足过滤条件的数据会被选择出来。并且 WHERE 可以结合 IN、NOT IN 联合使用。举个例子&＃xff1a;

SELECT name, age
FROM Table
WHERE name IN (SELECT name FROM Table2)


DISTINCT&＃xff1a;

DISTINCT 用于从数据集/流中去重根据 SELECT 的结果进行去重。

示例&＃xff1a;

SELECT DISTINCT name FROM Table;

对于流式查询&＃xff0c;计算查询结果所需的 State 可能会无限增长&＃xff0c;用户需要自己控制查询的状态范围&＃xff0c;以防止状态过大。

GROUP BY&＃xff1a;

GROUP BY 是对数据进行分组操作。例如我们需要计算成绩明细表中&＃xff0c;每个学生的总分。

示例&＃xff1a;

SELECT name, SUM(score) as TotalScore FROM Table GROUP BY name;

UNION  和  UNION ALL&＃xff1a;

UNION 用于将两个结果集合并起来&＃xff0c;要求两个结果集字段完全一致&＃xff0c;包括字段类型、字段顺序。不同于 UNION ALL 的是&＃xff0c;UNION 会对结果数据去重。

示例&＃xff1a;

SELECT * FROM T1 UNION (ALL) SELECT * FROM T2;

JOIN&＃xff1a;

JOIN 用于把来自两个表的数据联合起来形成结果表&＃xff0c;Flink 支持的 JOIN 类型包括&＃xff1a;

JOIN - INNER JOIN

LEFT JOIN - LEFT OUTER JOIN

RIGHT JOIN - RIGHT OUTER JOIN

FULL JOIN - FULL OUTER JOIN

这里的 JOIN 的语义和我们在关系型数据库中使用的 JOIN 语义一致。

示例&＃xff1a;

JOIN&＃xff08;将订单表数据和商品表进行关联&＃xff09;

SELECT * FROM Orders INNER JOIN Product ON Orders.productId &＃61; Product.id

LEFT JOIN 与 JOIN 的区别是当右表没有与左边相 JOIN 的数据时候&＃xff0c;右边对应的字段补 NULL 输出&＃xff0c;RIGHT JOIN 相当于 LEFT JOIN 左右两个表交互一下位置。FULL JOIN 相当于 RIGHT JOIN 和 LEFT JOIN 之后进行 UNION ALL 操作。

示例&＃xff1a;

SELECT * FROM Orders LEFT JOIN Product ON Orders.productId &＃61; Product.id
SELECT * FROM Orders RIGHT JOIN Product ON Orders.productId &＃61; Product.id
SELECT * FROM Orders FULL OUTER JOIN Product ON Orders.productId &＃61; Product.id


Group Window&＃xff1a;

根据窗口数据划分的不同&＃xff0c;目前 Apache Flink 有如下 3 种 Bounded Window&＃xff1a;

Tumble&＃xff0c;滚动窗口&＃xff0c;窗口数据有固定的大小&＃xff0c;窗口数据无叠加&＃xff1b;

Hop&＃xff0c;滑动窗口&＃xff0c;窗口数据有固定大小&＃xff0c;并且有固定的窗口重建频率&＃xff0c;窗口数据有叠加&＃xff1b;

Session&＃xff0c;会话窗口&＃xff0c;窗口数据没有固定的大小&＃xff0c;根据窗口数据活跃程度划分窗口&＃xff0c;窗口数据无叠加。

Tumble Window&＃xff1a;

Tumble 滚动窗口有固定大小&＃xff0c;窗口数据不重叠&＃xff0c;具体语义如下&＃xff1a;

Tumble 滚动窗口对应的语法如下&＃xff1a;

SELECT
    [gk],
    [TUMBLE_START(timeCol, size)],
    [TUMBLE_END(timeCol, size)],
    agg1(col1),
    ...
    aggn(colN)
FROM Tab1
GROUP BY [gk], TUMBLE(timeCol, size)


其中&＃xff1a;

[gk] 决定了是否需要按照字段进行聚合&＃xff1b;

TUMBLE_START 代表窗口开始时间&＃xff1b;

TUMBLE_END 代表窗口结束时间&＃xff1b;

timeCol 是流表中表示时间字段&＃xff1b;

size 表示窗口的大小&＃xff0c;如 秒、分钟、小时、天。

举个例子&＃xff0c;假如我们要计算每个人每天的订单量&＃xff0c;按照 user 进行聚合分组&＃xff1a;

SELECT user,
      TUMBLE_START(rowtime, INTERVAL ‘1’ DAY) as wStart,
      SUM(amount)
FROM Orders
GROUP BY TUMBLE(rowtime, INTERVAL ‘1’ DAY), user;


Hop Window&＃xff1a;

Hop 滑动窗口和滚动窗口类似&＃xff0c;窗口有固定的 size&＃xff0c;与滚动窗口不同的是滑动窗口可以通过 slide 参数控制滑动窗口的新建频率。因此当 slide 值小于窗口 size 的值的时候多个滑动窗口会重叠&＃xff0c;具体语义如下&＃xff1a;

Hop 滑动窗口对应语法如下&＃xff1a;

SELECT
    [gk],
    [HOP_START(timeCol, slide, size)] ,
    [HOP_END(timeCol, slide, size)],
    agg1(col1),
    ...
    aggN(colN)
FROM Tab1
GROUP BY [gk], HOP(timeCol, slide, size)


每次字段的意思和 Tumble 窗口类似&＃xff1a;

[gk] 决定了是否需要按照字段进行聚合&＃xff1b;

HOP_START 表示窗口开始时间&＃xff1b;

HOP_END 表示窗口结束时间&＃xff1b;

timeCol 表示流表中表示时间字段&＃xff1b;

slide 表示每次窗口滑动的大小&＃xff1b;

size 表示整个窗口的大小&＃xff0c;如 秒、分钟、小时、天。

举例说明&＃xff0c;我们要每过一小时计算一次过去 24 小时内每个商品的销量&＃xff1a;

SELECT product,
      SUM(amount)
FROM Orders
GROUP BY HOP(rowtime, INTERVAL &＃39;1&＃39; HOUR, INTERVAL &＃39;1&＃39; DAY), product


Session Window&＃xff1a;

会话时间窗口没有固定的持续时间&＃xff0c;但它们的界限由 interval 不活动时间定义&＃xff0c;即如果在定义的间隙期间没有出现事件&＃xff0c;则会话窗口关闭。

Seeeion 会话窗口对应语法如下&＃xff1a;

SELECT
    [gk],
    SESSION_START(timeCol, gap) AS winStart,
    SESSION_END(timeCol, gap) AS winEnd,
    agg1(col1),
     ...
    aggn(colN)
FROM Tab1
GROUP BY [gk], SESSION(timeCol, gap)


[gk] 决定了是否需要按照字段进行聚合&＃xff1b;

SESSION_START 表示窗口开始时间&＃xff1b;

SESSION_END 表示窗口结束时间&＃xff1b;

timeCol 表示流表中表示时间字段&＃xff1b;

gap 表示窗口数据非活跃周期的时长。

例如&＃xff0c;我们需要计算每个用户访问时间 12 小时内的订单量&＃xff1a;

SELECT user,
      SESSION_START(rowtime, INTERVAL ‘12’ HOUR) AS sStart,
      SESSION_ROWTIME(rowtime, INTERVAL ‘12’ HOUR) AS sEnd,
      SUM(amount)
FROM Orders
GROUP BY SESSION(rowtime, INTERVAL ‘12’ HOUR), user


Table API 和 SQL 捆绑在 flink-table Maven 工件中。必须将以下依赖项添加到你的项目才能使用 Table API 和 SQL&＃xff1a;

 
    org.apache.flink
    flink-table_2.11
    $flink.version



另外&＃xff0c;你需要为 Flink 的 Scala 批处理或流式 API 添加依赖项。对于批量查询&＃xff0c;您需要添加&＃xff1a;


  org.apache.flink
flink-scala_2.11
  $flink.version



2. Flink SQL 实战案例

1) 批数据 SQL

用法&＃xff1a;

1. 构建 Table 运行环境

2. 将 DataSet 注册为一张表

3. 使用 Table 运行环境的 sqlQuery 方法来执行 SQL 语句

示例&＃xff1a;使用 Flink SQL 统计用户消费订单的总金额、最大金额、最小金额、订单总数。

测试数据&＃xff08;订单 ID、用户名、订单日期、订单金额&＃xff09;:

Order(1, "zhangsan", "2018-10-20 15:30", 358.5),
Order(2, "zhangsan", "2018-10-20 16:30", 131.5),
Order(3, "lisi", "2018-10-20 16:30", 127.5),
Order(4, "lisi", "2018-10-20 16:30", 328.5),
Order(5, "lisi", "2018-10-20 16:30", 432.5),
Order(6, "zhaoliu", "2018-10-20 22:30", 451.0),
Order(7, "zhaoliu", "2018-10-20 22:30", 362.0),
Order(8, "zhaoliu", "2018-10-20 22:30", 364.0),
Order(9, "zhaoliu", "2018-10-20 22:30", 341.0)


步骤:

1. 获取一个批处理运行环境

2. 获取一个 Table 运行环境

3. 创建一个样例类 Order 用来映射数据&＃xff08;订单名、用户名、订单日期、订单金额&＃xff09;

4. 基于本地 Order 集合创建一个 DataSet source

5. 使用 Table 运行环境将 DataSet 注册为一张表

6. 使用 SQL 语句来操作数据&＃xff08;统计用户消费订单的总金额、最大金额、最小金额、订单总数&＃xff09;

7. 使用 TableEnv.toDataSet 将 Table 转换为 DataSet

8. 打印测试

示例代码&＃xff1a;

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.table.api.Table, TableEnvironment
import org.apache.flink.table.api.scala.BatchTableEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.types.Row
/**
 * 使用Flink SQL统计用户消费订单的总金额、最大金额、最小金额、订单总数。
 */
object BatchFlinkSqlDemo 
  //3. 创建一个样例类 Order 用来映射数据&＃xff08;订单名、用户名、订单日期、订单金额&＃xff09;
  case class Order(id:Int, userName:String, createTime:String, money:Double)
  def main(args: Array[String]): Unit &＃61; 
    /**
     * 实现思路&＃xff1a;
     * 1. 获取一个批处理运行环境
     * 2. 获取一个Table运行环境
     * 3. 创建一个样例类 Order 用来映射数据&＃xff08;订单名、用户名、订单日期、订单金额&＃xff09;
     * 4. 基于本地 Order 集合创建一个DataSet source
     * 5. 使用Table运行环境将DataSet注册为一张表
     * 6. 使用SQL语句来操作数据&＃xff08;统计用户消费订单的总金额、最大金额、最小金额、订单总数&＃xff09;
     * 7. 使用TableEnv.toDataSet将Table转换为DataSet
     * 8. 打印测试
     */
    //1. 获取一个批处理运行环境
    val env: ExecutionEnvironment &＃61; ExecutionEnvironment.getExecutionEnvironment
    //2. 获取一个Table运行环境
    val tabEnv: BatchTableEnvironment &＃61; TableEnvironment.getTableEnvironment(env)
    //4. 基于本地 Order 集合创建一个DataSet source
    val orderDataSet: DataSet[Order] &＃61; env.fromElements(
      Order(1, "zhangsan", "2018-10-20 15:30", 358.5),
      Order(2, "zhangsan", "2018-10-20 16:30", 131.5),
      Order(3, "lisi", "2018-10-20 16:30", 127.5),
      Order(4, "lisi", "2018-10-20 16:30", 328.5),
      Order(5, "lisi", "2018-10-20 16:30", 432.5),
      Order(6, "zhaoliu", "2018-10-20 22:30", 451.0),
      Order(7, "zhaoliu", "2018-10-20 22:30", 362.0),
      Order(8, "zhaoliu", "2018-10-20 22:30", 364.0),
      Order(9, "zhaoliu", "2018-10-20 22:30", 341.0)
    )
    //5. 使用Table运行环境将DataSet注册为一张表
    tabEnv.registerDataSet("t_order", orderDataSet)
    //6. 使用SQL语句来操作数据&＃xff08;统计用户消费订单的总金额、最大金额、最小金额、订单总数&＃xff09;
    //用户消费订单的总金额、最大金额、最小金额、订单总数。
    val sql &＃61;
      """
        | select
        |   userName,
        |   sum(money) totalMoney,
        |   max(money) maxMoney,
        |   min(money) minMoney,
        |   count(1) totalCount
        |  from t_order
        |  group by userName
        |""".stripMargin  //在scala中stripMargin默认是“|”作为多行连接符
    //7. 使用TableEnv.toDataSet将Table转换为DataSet
    val table: Table &＃61; tabEnv.sqlQuery(sql)
    table.printSchema()
    tabEnv.toDataSet[Row](table).print()
  


2) 流数据 SQL

流处理中也可以支持 SQL。但是需要注意以下几点&＃xff1a;

1. 要使用流处理的 SQL&＃xff0c;必须要添加水印时间

2. 使用 registerDataStream 注册表的时候&＃xff0c;使用 &＃39; 来指定字段

3. 注册表的时候&＃xff0c;必须要指定一个 rowtime&＃xff0c;否则无法在 SQL 中使用窗口

4. 必须要导入 import org.apache.flink.table.api.scala._ 隐式参数

5. SQL 中使用 trumble(时间列名, interval &＃39;时间&＃39; sencond) 来进行定义窗口

示例&＃xff1a;使用 Flink SQL 来统计 5 秒内 用户的 订单总数、订单的最大金额、订单的最小金额。

步骤

1. 获取流处理运行环境

2. 获取 Table 运行环境

3. 设置处理时间为 EventTime

4. 创建一个订单样例类 Order &＃xff0c;包含四个字段&＃xff08;订单 ID、用户 ID、订单金额、时间戳&＃xff09;

5. 创建一个自定义数据源
   

   • 使用 for 循环生成 1000 个订单

   • 随机生成订单 ID&＃xff08;UUID&＃xff09;

   • 随机生成用户 ID&＃xff08;0-2&＃xff09;

   • 随机生成订单金额&＃xff08;0-100&＃xff09;

   • 时间戳为当前系统时间

   • 每隔 1 秒生成一个订单

6. 添加水印&＃xff0c;允许延迟 2 秒

7. 导入 import org.apache.flink.table.api.scala._ 隐式参数

8. 使用 registerDataStream 注册表&＃xff0c;并分别指定字段&＃xff0c;还要指定 rowtime 字段

9. 编写 SQL 语句统计用户订单总数、最大金额、最小金额 分组时要使用 tumble(时间列, interval &＃39;窗口时间&＃39; second) 来创建窗口

10. 使用 tableEnv.sqlQuery 执行 sql 语句

11. 将 SQL 的执行结果转换成 DataStream 再打印出来

12. 启动流处理程序

示例代码&＃xff1a;

import java.util.UUID
import java.util.concurrent.TimeUnit
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.source.RichSourceFunction, SourceFunction
import org.apache.flink.streaming.api.scala.DataStream, StreamExecutionEnvironment
import org.apache.flink.table.api.Table, TableEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.types.Row
import scala.util.Random
/**
 * 需求&＃xff1a;
 *  使用Flink SQL来统计5秒内 用户的 订单总数、订单的最大金额、订单的最小金额
 *
 *  timestamp是关键字不能作为字段的名字&＃xff08;关键字不能作为字段名字&＃xff09;
 */
object StreamFlinkSqlDemo 
    /**
     *  1. 获取流处理运行环境
     * 2. 获取Table运行环境
     * 3. 设置处理时间为 EventTime
     * 4. 创建一个订单样例类 Order &＃xff0c;包含四个字段&＃xff08;订单ID、用户ID、订单金额、时间戳&＃xff09;
     * 5. 创建一个自定义数据源
     *    使用for循环生成1000个订单
     *    随机生成订单ID&＃xff08;UUID&＃xff09;
     *    随机生成用户ID&＃xff08;0-2&＃xff09;
     *    随机生成订单金额&＃xff08;0-100&＃xff09;
     *    时间戳为当前系统时间
     *    每隔1秒生成一个订单
     * 6. 添加水印&＃xff0c;允许延迟2秒
     * 7. 导入 import org.apache.flink.table.api.scala._ 隐式参数
     * 8. 使用 registerDataStream 注册表&＃xff0c;并分别指定字段&＃xff0c;还要指定rowtime字段
     * 9. 编写SQL语句统计用户订单总数、最大金额、最小金额
     * 分组时要使用 tumble(时间列, interval &＃39;窗口时间&＃39; second) 来创建窗口
     * 10. 使用 tableEnv.sqlQuery 执行sql语句
     * 11. 将SQL的执行结果转换成DataStream再打印出来
     * 12. 启动流处理程序
     */
    // 3. 创建一个订单样例类&＃96;Order&＃96;&＃xff0c;包含四个字段&＃xff08;订单ID、用户ID、订单金额、时间戳&＃xff09;
    case class Order(orderId:String, userId:Int, money:Long, createTime:Long)
    def main(args: Array[String]): Unit &＃61; 
      // 1. 创建流处理运行环境
      val env: StreamExecutionEnvironment &＃61; StreamExecutionEnvironment.getExecutionEnvironment
      // 2. 设置处理时间为&＃96;EventTime&＃96;
      env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
      //获取table的运行环境
      val tableEnv &＃61; TableEnvironment.getTableEnvironment(env)
      // 4. 创建一个自定义数据源
      val orderDataStream &＃61; env.addSource(new RichSourceFunction[Order] 
        var isRunning &＃61; true
        override def run(ctx: SourceFunction.SourceContext[Order]): Unit &＃61; 
          // - 随机生成订单ID&＃xff08;UUID&＃xff09;
          // - 随机生成用户ID&＃xff08;0-2&＃xff09;
          // - 随机生成订单金额&＃xff08;0-100&＃xff09;
          // - 时间戳为当前系统时间
          // - 每隔1秒生成一个订单
          for (i <- 0 until 1000 if isRunning) 
            val order &＃61; Order(UUID.randomUUID().toString, Random.nextInt(3), Random.nextInt(101),
              System.currentTimeMillis())
            TimeUnit.SECONDS.sleep(1)
            ctx.collect(order)
          
        
        override def cancel(): Unit &＃61;  isRunning &＃61; false 
      )
      // 5. 添加水印&＃xff0c;允许延迟2秒
      val watermarkDataStream &＃61; orderDataStream.assignTimestampsAndWatermarks(
        new BoundedOutOfOrdernessTimestampExtractor[Order](Time.seconds(2)) 
          override def extractTimestamp(element: Order): Long &＃61; 
            val eventTime &＃61; element.createTime
            eventTime
          
        
      )
      // 6. 导入&＃96;import org.apache.flink.table.api.scala._&＃96;隐式参数
      // 7. 使用&＃96;registerDataStream&＃96;注册表&＃xff0c;并分别指定字段&＃xff0c;还要指定rowtime字段
      import org.apache.flink.table.api.scala._
      tableEnv.registerDataStream("t_order", watermarkDataStream, &＃39;orderId, &＃39;userId, &＃39;money,&＃39;createTime.rowtime)
      // 8. 编写SQL语句统计用户订单总数、最大金额、最小金额
      // - 分组时要使用&＃96;tumble(时间列, interval &＃39;窗口时间&＃39; second)&＃96;来创建窗口
      val sql &＃61;
      """
        |select
        | userId,
        | count(1) as totalCount,
        | max(money) as maxMoney,
        | min(money) as minMoney
        | from
        | t_order
        | group by
        | tumble(createTime, interval &＃39;5&＃39; second),
        | userId
      """.stripMargin
      // 9. 使用&＃96;tableEnv.sqlQuery&＃96;执行sql语句
      val table: Table &＃61; tableEnv.sqlQuery(sql)
      // 10. 将SQL的执行结果转换成DataStream再打印出来
      table.toRetractStream[Row].print()
      env.execute("StreamSQLApp")